package com.dynamictpm.service;

import com.dynamictpm.config.ConfigExecutor;
import com.dynamictpm.notify.Event;
import com.dynamictpm.notify.LocalDataChangeEvent;
import com.dynamictpm.notify.NotifyCenter;
import com.dynamictpm.notify.Subscriber;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 描述：： 长连接服务类
 * <p>
 * ------------------------------------
 *
 * @program: dynamic-tpm
 * @ClassName LongPollingService
 * @author: it-yml
 * @create: 2023-07-05 14:16
 * @Version 1.0
 * --------------------------------------
 **/
@Slf4j
@Service
public class LongPollingService {


    final Queue<ClientLongPolling> allSubs;
    public LongPollingService() {
        this.allSubs = new ConcurrentLinkedQueue();
        ConfigExecutor.scheduleLongPolling(new StatTask(), 0L, 10L, TimeUnit.SECONDS);
        NotifyCenter.registerSubscriber(new Subscriber() {

            @Override
            public void onEvent(Event event) {
//                if (isFixedPolling()) {
//                    // Ignore.
//                } else {
                    if (event instanceof LocalDataChangeEvent) {
                        LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                        ConfigExecutor.executeLongPolling(new DataChangeTask(evt.groupKey));
//                    }
                }
            }

            @Override
            public Class<? extends Event> subscribeType() {
                return LocalDataChangeEvent.class;
            }
        });
    }

    public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp
//            , Map<String, String> clientMd5Map,
//                                     int probeRequestSize
    ) {
//        String str = req.getHeader(LONG_POLLING_HEADER);
//        String appName = req.getHeader(CLIENT_APPNAME_HEADER);
//        int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
//
//        long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
//        if (isFixedPolling()) {
//            timeout = Math.max(10000, getFixedPollingInterval());
//        } else {
//            List<String> changedGroups = Md5ConfigUtil.compareMd5(req, clientMd5Map);
//            if (changedGroups.size() > 0) {
//                generateResponse(rsp, changedGroups);
//                return;
//            }
//        }
//
//        String ip = RequestUtil.getRemoteIp(req);

        final AsyncContext asyncContext = req.startAsync();
        asyncContext.setTimeout(0L);

        ConfigExecutor.executeLongPolling(new ClientLongPolling(asyncContext,  29500l, "测试app"));
    }


    class StatTask implements Runnable {

        @Override
        public void run() {
            log.info("[long-pulling] client count " + allSubs.size());
        }
    }
    class DataChangeTask implements Runnable{
        final String groupKey;

        DataChangeTask(String groupKey) {
            this.groupKey = groupKey;
        }

        @Override
        public void run() {
            try {
                for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
                    ClientLongPolling clientSub = iter.next();

//                    if (clientSub.clientMd5Map.containsKey(groupKey)) {
//                        getRetainIps().put(clientSub.ip, System.currentTimeMillis());
//                        ConfigCacheService.updateMd5(groupKey, ConfigCacheService.getContentMd5(groupKey), System.currentTimeMillis());
//                        iter.remove();
                        clientSub.sendResponse(Arrays.asList(groupKey));
//                    }
                }
            } catch (Exception ex) {
                log.error("Data change error :: {}", ex.getMessage(), ex);
            }
        }
    }

    class ClientLongPolling implements Runnable {

        final long createTime;

        final AsyncContext asyncContext;

        final String appName;


        final long timeoutTime;

        Future<?> asyncTimeoutFuture;



        public ClientLongPolling(AsyncContext asyncContext, long timeout, String appName) {
            this.asyncContext = asyncContext;
            this.timeoutTime = timeout;
            this.appName = appName;
            this.createTime = System.currentTimeMillis();
        }

        @Override
        public void run() {
            asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(() -> {
                try {
                    allSubs.remove(ClientLongPolling.this);
                    System.out.println("轮询");
                    sendResponse(null);
                } catch (Exception ex) {

                }

            }, timeoutTime, TimeUnit.MILLISECONDS);
            allSubs.add(this);
        }

        private void sendResponse(List<String> changedGroups) {

            // Cancel time out task.
            if (null != asyncTimeoutFuture) {
                asyncTimeoutFuture.cancel(false);
            }
            generateResponse(changedGroups);
        }

        private void generateResponse(List<String> changedGroups) {

            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();

            try {

                String resultStr = "success";

                response.setHeader("Pragma", "no-cache");
                response.setDateHeader("Expires", 0);
                response.setHeader("Cache-Control", "no-cache,no-store");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println(resultStr);
                asyncContext.complete();
            } catch (Exception ex) {

                asyncContext.complete();
            }
        }

    }

}
